package com.mq.redis.queue.executor;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Controller;

import com.mq.model.WorksPoolConfig;

/**
 * redis队列处理线程池
 * 
 * @author wenfei
 *
 */
@Controller
public class RedisQueueWorkerExecutor {

	/**
	 * 
	 * 1. 维护两个队列：pending队列和doing表(hash表). 2. workers定义为ThreadPool。 3.
	 * 由pending队列出队后，workers分配一个线程（单个worker）去处理消息——给目标消息append一个当前时间戳和当前线程名称，
	 * 将其写入doing表，然后该worker去消费消息，完成后自行在doing表擦除信息。 4.
	 * 启用一个定时任务，每隔一段时间去扫描doing队列，检查每隔元素的时间戳，如果超时，
	 * 则由worker的ThreadPoolExecutor去检查线程是否存在，如果存在则取消当前任务执行，
	 * 并把事务rollback。最后把该任务从doing队列中pop出，再重新push进pending队列。 5.
	 * 在worker的某线程中，如果处理业务失败，则主动回滚，并把任务从doing队列中移除，重新push进pending队列。
	 * 
	 */

	public ExecutorService executorService;

	public RedisQueueWorkerExecutor() {
		// 默认构造采用缓存线程池
		this.executorService = Executors.newCachedThreadPool();
	}

	public RedisQueueWorkerExecutor(WorksPoolConfig config) {
		if (config.isCustom()) {
			BlockingQueue<Runnable> queue = null;
			if (config.getWorkQueue() > 0) {
				queue = new LinkedBlockingQueue<Runnable>(config.getWorkQueue()); // 可能根据机器jvm配置定义
																					
			} else {
				// 没有配置情况下, 默认采用有序无界队列
				queue = new LinkedBlockingQueue<Runnable>();
			}
			// 配置线程池信息
			this.executorService = new ThreadPoolExecutor(
					// 默认大小                         // 最大连接数 
					config.getCorePoolSize(), config.getMaximumPoolSize(),
					// 回收时间                                           // 回收时间单位            // 采用队列
					config.getKeepAliveTime(), TimeUnit.SECONDS, queue,
					// 拒绝策略，任务队列满后，新的任务将被丢弃，并抛出异常
					new ThreadPoolExecutor.AbortPolicy()
			);
		} else {
			this.executorService = Executors.newCachedThreadPool();
		}
	}

	/*
	 * 创建线程，对线程处理事件
	 */
	public void execute(Runnable runnable) {
		executorService.execute(runnable);
	}

	


		
}
